
#include "stdio.h"
//#include "windows.h"
#include "comm/comm.h"
#include "comm/queue.h"
#include "comm/comm_frame.h"
#include "comm/comm_socket.h"
#include "comm/thread.h"

static mutex_t m_send_mutex;
static thread_t m_send_thread_fd;
static int send_thread_on = 1;

static ssem_t m_send_semaphore;
static queue_list m_send_queue;
static comm_frame_send_context_t m_send_frame_ctx;

#ifdef COMM_RECV_ENABLE

#ifdef WIN32

thread_t m_recv_thread_fd = NULL;
static int recv_thread_on = 1;

static HANDLE m_recv_semaphore = NULL;
static queue_list m_recv_queue;
static comm_frame_recv_context_t m_recv_frame_ctx;

static thread_t m_recv_callback_thread_fd = NULL;
static int recv_callback_thread_on = 1;

#else

thread_t m_recv_thread_fd;
static int recv_thread_on = 1;

static ssem_t m_recv_semaphore;
static queue_list m_recv_queue;
static comm_frame_recv_context_t m_recv_frame_ctx;

static thread_t m_recv_callback_thread_fd;
static int recv_callback_thread_on = 1;

#endif /* ifdef WIN32 */

#endif /* ifdef COM_RECV_ENABLE */

int comm_send(void* fd, uint8_t* buf, int buf_len)
{	
	comm_packet_t* packet = NULL;

	//if (mutex_trylock(&m_send_mutex))
	{
		packet = malloc(sizeof(char)*(buf_len + COMM_PACKET_HEADER_SIZE));

		if ( packet == NULL)
		{
			TRACE("comm_send malloc fail !!!");
			return -1;
		}

		packet->len = buf_len;	
		memcpy(packet->data, buf, buf_len);
		enqueue(&m_send_queue, (void*)packet);
		ssem_post(&m_send_semaphore);
	}
	
	//mutex_unlock(&m_send_mutex);
}

int comm_do_send()
{
	comm_packet_t* packet = NULL;
	int ret = 0;

	ret = dequeue(&m_send_queue, (void**)&packet);

	if( ret < 0 && packet == NULL ) 
	{
		return -1;
	}

	comm_frame_send(&m_send_frame_ctx, packet->data, packet->len);

	free(packet);
}

void comm_send_thread(void* arg)
{
	send_thread_on = 1;
	
	m_send_frame_ctx.state = COMM_FRAME_SEND_READY;
	m_send_frame_ctx.upd_fd = udp_create();

	m_send_frame_ctx.dest_addr.sin_family=AF_INET;
	m_send_frame_ctx.dest_addr.sin_addr.s_addr = app_global.config.send_ip;
	m_send_frame_ctx.dest_addr.sin_port= app_global.config.send_port;

	while(send_thread_on)
	{
		ssem_wait(&m_send_semaphore);

		//TRACE("\r\ncomm_do_send");
		comm_do_send();
	}
}

#ifdef COMM_RECV_ENABLE

void comm_recv_done_cb()
{
	enqueue(&m_recv_queue, (void*)m_recv_frame_ctx.recv_buf);
	m_recv_frame_ctx.recv_buf = NULL;

    ssem_post(&m_recv_semaphore);
}

void comm_recv_thread(void* arg)
{
	static uint8_t recv_thread_buf[COMM_FRAME_PACKET_MAX_LEN * 2];	
	int len = 0;
	int ret = 0;

	recv_thread_on = 1;

	m_recv_frame_ctx.state = COMM_FRAME_RECV_READY;
	m_recv_frame_ctx.recv_buf = NULL;

	m_recv_frame_ctx.upd_fd = udp_create();

	memset(&m_recv_frame_ctx.dest_addr, 0, sizeof(m_recv_frame_ctx.dest_addr));
	m_recv_frame_ctx.dest_addr_len = sizeof(m_recv_frame_ctx.dest_addr);
	
	m_recv_frame_ctx.local_addr.sin_family= AF_INET;
	m_recv_frame_ctx.local_addr.sin_port= app_global.config.recv_port; 
	m_recv_frame_ctx.local_addr.sin_addr.s_addr = INADDR_ANY; 

	ret = bind( m_recv_frame_ctx.upd_fd, &m_recv_frame_ctx.local_addr, sizeof(m_recv_frame_ctx.local_addr));

	if ( ret < 0 )
	{
		TRACE("bind error !!! ");
		exit(1);
	}

	while(recv_thread_on)
	{
		
		len = comm_frame_do_recv(&m_recv_frame_ctx, recv_thread_buf, sizeof(recv_thread_buf));

		if ( len < 0)
		{
			TRACE("\r\n recv failed\n");
		}
		else if ( len == 0)
		{
			TRACE("\r\nconnect close!!");
		}
		else
		{
			comm_frame_recv(&m_recv_frame_ctx,recv_thread_buf,len);
		}
	}
}

void comm_recv_callback_thread(void *arg)
{
	static comm_packet_t *packet = NULL;

	recv_callback_thread_on = 1;
	
	while(recv_callback_thread_on)
	{
		ssem_wait(&m_recv_semaphore);

		if ( dequeue(&m_recv_queue, (void**)&packet) )
		{
			if ( app_global.config.pfn_recv_cb)
			{
				app_global.config.pfn_recv_cb(NULL, packet->data, packet->len);
			}

			free(packet);
			packet = NULL;
		}
	}
}

#else
void comm_recv_done_cb(){}

#endif /* ifdef COMM_RECV_ENABLE */


int comm_initialize()
{
	socket_init();

	if ( app_global.config.flags & COMM_LIB_FLAGS_SEND_ENABLE)
	{
		thread_create(&m_send_thread_fd, comm_send_thread);
        mutex_init(&m_send_mutex);
    	ssem_init(&m_send_semaphore);
    	queue_init(&m_send_queue);    
	}
	
#ifdef COMM_RECV_ENABLE
    if(app_global.config.flags & COMM_LIB_FLAGS_RECV_ENABLE)
    {   
        thread_create(&m_recv_thread_fd, comm_recv_thread);
        ssem_init(&m_recv_semaphore);
        queue_init(&m_recv_queue);
        thread_create(&m_recv_callback_thread_fd, comm_recv_callback_thread);
    }

#endif /* ifdef COMM_RECV_ENABLE */

	return 0;
}

int comm_poll()
{
	return 0;
}

int comm_finalize()
{
	if (m_send_thread_fd != NULL)
	{
		thread_destroy(&m_send_thread_fd);    
    	ssem_destroy(&m_send_semaphore);
        mutex_destroy(&m_send_mutex);
	}

#ifdef COMM_RECV_ENABLE

    thread_destroy(&m_recv_thread_fd);
    thread_destroy(&m_recv_callback_thread_fd);
    ssem_destroy(&m_recv_semaphore);
    
#endif /* ifdef COMM_RECV_ENBALE */

	return 0;
}


